package com.atguigu.flink.exec1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.flink.function.MySourceFunction;
import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Set;

/**

 */
public class Demo3_Channel
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                env
                   .addSource(new MySourceFunction()).setParallelism(1)
                   .map(new MapFunction<String, Tuple2<String,Integer>>()
                   {
                       @Override
                       public Tuple2<String,Integer> map(String jsonStr) throws Exception {

                           //map  { userId:100,behavior:"uninstall","channel":"ZTE",timestamp:xxx  }
                           JSONObject jsonObject = JSON.parseObject(jsonStr);

                           return Tuple2.of(jsonObject.getString("channel") + "_" +jsonObject.getString("behavior"),1);

                       }
                   })
                   .keyBy(t -> t.f0)
                   .sum(1)
                   .print().setParallelism(1);

        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
